package com.atguigu.chapter11;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @ClassName: Flink14_Function_Scalar
 * @Description:
 * @Author: kele
 * @Date: 2021/4/14 0:09
 *
 * 自定义scalar函数(标量函数，一进一出)
 *  1、
 *      -- 输入: 0个1个或多个
 *      -- 输出: 1个
 *
 *  2、在类中必须实现 eval的方法
 *
 *  table：
 *      1、内联方式
 *          call(xxx.class,参数)
 *      2、注册方式
 *         .createTemporaryFunction("自定义一个函数名",xxx.class);
 *         call(xxx,参数)
 *  SQL：
 *      先注册， select xxx(参数) from xxx
 *
 **/
//将小写转化为大写
public class Flink14_Function__Scalar {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        DataStreamSource<String> stream = env.fromElements("hello", "atguigu", "Hello");

        Table table = tEnv.fromDataStream(stream, $("words"));

    /**
     *
     * 在Table的API中使用
     */
        //方式一、内联方式，直接使用
        /*table.select($("words"),call(MyFunction.class,$("words").as("UPPER")))
                .execute()
                .print();*/

        //方式二、注册方式，先将函数注册，再使用
   /*     tEnv.createTemporaryFunction("toUpper",MyFunction.class);

        table.select($("words"),call("toUpper",$("words")).as("UPPER") )
                .execute()
                .print();*/



    /**
     *
     * 在SQL中使用
     *
     * 在sql中使用必须先注册
     */

    tEnv.createTemporaryView("t_word",table);

        tEnv.createTemporaryFunction("toUpper",MyFunction.class);

        tEnv.sqlQuery("select words,toUpper(words) w_Upper from t_word")
                .execute().print();


    }

    /**
     * 自定义函数；将小写字母转化为大写
     *
     * 在自定义的函数中必须实现一个eval的方法
     */
    public static class MyFunction extends ScalarFunction {

        //自定义的eval函数，在使用时会自动调用，使用时会直接将参数传入，并将该函数的返回值返回
        public String eval(String line){
            return line.toUpperCase();
        }

    }

}
